package org.ws.task.netty.handler;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.TaskMainBootstrap;
import org.ws.task.exception.NoItemProcessorException;
import org.ws.task.exception.RequestHandlerException;
import org.ws.task.item.AbstractTask;
import org.ws.task.item.ExecuteContext;
import org.ws.task.item.ItemWorkerProcessor;
import org.ws.task.item.RemoteAccessApi;
import org.ws.task.item.TaskInstance;
import org.ws.task.listener.TaskItemListenerManager;
import org.ws.task.repository.NodeMetadataRepository;
import org.ws.task.repository.TaskRepository;
import org.ws.task.model.protocol.CommitRequest;
import org.ws.task.model.protocol.PushItemRequest;
import org.ws.task.model.protocol.PushItemResponse;
import org.ws.task.model.TaskItem;
import org.ws.task.netty.RequestActionHandler;
import org.ws.task.model.protocol.base.RequestAction;
import org.ws.task.model.protocol.base.RequestMessage;
import org.ws.task.model.protocol.base.ResponseMessage;
import org.ws.task.model.protocol.base.ResponseStatus;
import org.ws.task.model.TaskProcessStatus;
import org.ws.task.repository.WorkerContextRepository;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public class PushItemWorkerHandler implements RequestActionHandler<PushItemRequest<TaskItem>, PushItemResponse> {

    private static Logger log = LoggerFactory.getLogger(PushItemWorkerHandler.class.getName());

    private final ExecutorService processExecutor;
    private final Map<String,ItemWorkerProcessorThread> processorThreadMap;

    public PushItemWorkerHandler(ExecutorService processExecutor) {
        this.processExecutor = processExecutor;
        this.processorThreadMap = Maps.newConcurrentMap();
    }

    @Override
    public RequestAction action() {
        return RequestAction.PUSH_TASK;
    }

    @Override
    public ResponseMessage<PushItemResponse> handler(ChannelHandlerContext ctx, RequestMessage<PushItemRequest<TaskItem>> requestMessage) throws RequestHandlerException {
        // 接收数据加入执行队列，并发送执行
        PushItemRequest pushItemRequest = requestMessage.getBodyObject();
        String taskName = pushItemRequest.getTaskName();
        Collection<TaskItem> items = pushItemRequest.getItems();

        // 接收新任务到worker本地context
        final TaskInstance taskInstance = TaskRepository.getInstance().getTaskInstance(taskName);
        final ExecuteContext context = WorkerContextRepository.getInstance().getContext(taskName);
        context.addAll(items);
        // 执行:每个task一个线程
        if(!this.processorThreadMap.containsKey(taskName)){
            ItemWorkerProcessorThread processorThread = new ItemWorkerProcessorThread(taskInstance,context);
            this.processorThreadMap.put(taskName,processorThread);
            this.processExecutor.submit(processorThread);
        }
        // response
        List<Long> ids = items.stream().map(TaskItem::getId).collect(Collectors.toList());
        PushItemResponse itemResponse = new PushItemResponse();
        itemResponse.setItemIds(ids);
        itemResponse.setSuccess(true);

        ResponseMessage<PushItemResponse> responseMessage = new ResponseMessage<>(requestMessage.getId());
        responseMessage.setBodyObject(itemResponse);
        responseMessage.setSuccess(true);
        responseMessage.setStatus(ResponseStatus.SUCCESS.getCode());
        return responseMessage;
    }

    private static class ItemWorkerProcessorThread implements Runnable {

        private final TaskInstance taskInstance;
        private final ExecuteContext context;
        private final AtomicBoolean running;
        private static final Timer processTimer = new Timer("PushItemWorkerHandlerTimer", false);

        public ItemWorkerProcessorThread(TaskInstance taskInstance,ExecuteContext context) {
            this.taskInstance = taskInstance;
            this.context = context;
            this.running = new AtomicBoolean(false);
        }

        @Override
        public void run() {
            // 每1秒执行一次是否存在新数据
            this.processTimer.scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    process();
                }
            }, 300, 1_000);
        }

        private void process(){
            if(this.running.get()){
                return;
            }
            this.running.set(true);
            String name = taskInstance.getTaskConfig().getName();
            try {
                ItemWorkerProcessor itemWorkerProcessor = taskInstance.getItemWorkerProcessor();
                if (itemWorkerProcessor == null) {
                    throw new NoItemProcessorException("task[{" + taskInstance.getTaskConfig().getName() + "}] itemProcessor is null.");
                }
                AbstractTask task = (AbstractTask) taskInstance.getTask();
                TaskItemListenerManager listenerManager = task.getTaskItemListenerManager();
                RemoteAccessApi remoteAccessApi = task.getRemoteAccessApi();
                String nodeId = NodeMetadataRepository.getInstance().getTheWorkerNode().getId();

                final Long timeout = taskInstance.getTaskConfig().getFetchTimeout();
                final Integer chunk = taskInstance.getTaskConfig().getChunk();
                AtomicInteger index = new AtomicInteger(0);
                List<TaskItem> items = Lists.newArrayList();
                while (!context.isEmpty()) {
                    final TaskItem taskItem = context.poll();
                    if (taskItem != null) {
                        int i = index.incrementAndGet();
                        items.add(taskItem);
                        if(i % chunk == 0){
                            doProcess(name,nodeId,items,itemWorkerProcessor,timeout,context,listenerManager,remoteAccessApi);
                            items = Lists.newArrayList();
                        }
                    }
                }
                doProcess(name,nodeId,items,itemWorkerProcessor,timeout,context,listenerManager,remoteAccessApi);
            }
            catch (Exception e){
                log.error(e.getMessage(),e);
            }
            finally {
                this.running.set(false);
            }
        }

        private void doProcess(String name, String nodeId, List<TaskItem> items, ItemWorkerProcessor itemWorkerProcessor, Long timeout, ExecuteContext context,
                               TaskItemListenerManager listenerManager, RemoteAccessApi remoteAccessApi){
            if(CollectionUtils.isEmpty(items)){
                return;
            }
            //
            try {
                listenerManager.beforeProcess(items);
                itemWorkerProcessor.process(items);
                listenerManager.afterProcess(items);
                // 提交状态
                remoteAccessApi.commitStatusSync(new CommitRequest(name,items,nodeId), timeout);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                for (TaskItem item : items) {
                    context.commitStatus(item.getId(), TaskProcessStatus.FAILED);
                }
            }
        }
    }


}
